﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using BDAlertorApp.Model;
using BDAlertorApp.Protocol;
using BDAlertorApp.Util;


namespace BDAlertorApp.Service
{
    /// <summary>
    /// 数据解析服务
    /// </summary>
    class DataPacketProcessService
    {
        private static log4net.ILog logger = log4net.LogManager.GetLogger("dataPacketProcessService");

        /// <summary>
        /// 接受数据队列
        /// </summary>
        private ConcurrentQueue<byte[]> recvQueue = new ConcurrentQueue<byte[]>();
        private ConcurrentQueue<byte[]> parseQueue = new ConcurrentQueue<byte[]>();

        private ByteBuffer byteBuffer = ByteBuffer.Allocate(1024);
        private int pOffset;

        private Boolean isContinue = false;
        private Thread processThread = null;
        private Thread parseThread = null;

        private static DataPacketProcessService instance = new DataPacketProcessService();

        public static DataPacketProcessService GetInstance()
        {
            return instance;
        }

        public void Start()
        {
            logger.Info("数据解析服务开启");
            isContinue = true;
            processThread = new Thread(new ThreadStart(ProcessThreadFunc));
            processThread.Start();
            parseThread = new Thread(new ThreadStart(ParseThreadFunc));
            parseThread.Start();
        }

        public void Stop()
        {
            logger.Info("数据解析服务关闭");
            try
            {
                isContinue = false;
                if (processThread != null)
                {
                    processThread.Abort();
                }
                if (parseThread != null)
                {
                    parseThread.Abort();
                }
            }
            catch (Exception ex)
            {
                logger.Error(ex.Message, ex);
                logger.Error(ex.StackTrace);
            }
        }

        public void ReadyToProcess(byte[] bytes)
        {
//            logger.Info("ReadyToProcess->" + Encoding.Default.GetString(bytes));

            recvQueue.Enqueue(bytes);
        }

        public event OnPacketNotifyEventHandler OnPacketNotify;
        public void PacketNotify(BdPacketData bdPacketData)
        {
            if (OnPacketNotify != null)
            {
                OnPacketNotify(bdPacketData);
            }
        }

        public event OnDlEventHandler OnDlNotify;
        public void DlNotify(byte[] buffer)
        {
            if (OnDlNotify != null)
            {
                OnDlNotify(buffer);
            }
        }

        /// <summary>
        /// 处理线程，取出完整的一条数据，并送入解析队列
        /// </summary>
        private void ProcessThreadFunc()
        {
            while (isContinue)
            {

                while (recvQueue.Count > 0)
                {
                    try
                    {
                        byte[] buffer;
                        recvQueue.TryDequeue(out buffer);
                        int len = buffer.Length;
                        for (int i = 0; i < len; i++)
                        {
                    
                            byteBuffer.WriteByte(buffer[i]);
                            if (buffer[i] == '\n')
                            {
                                byte[] data = byteBuffer.ToArray();
                                logger.Info("ProcessThreadFunc->完整协议：" + Encoding.Default.GetString(data));
                                DlNotify(data);
                                //重置byteBuffer 索引
                                byteBuffer.Reset();
                            }
                            
                        }
                    }
                    catch(Exception ex)
                    {
                        logger.Error(ex.Message, ex);
                        logger.Error(ex.StackTrace);
                    }
                }

                Thread.Sleep(10);
            }
        }

        /// <summary>
        /// 解析线程
        /// </summary>
        private void ParseThreadFunc()
        {
            while (isContinue)
            {
                List<byte[]> result = new List<byte[]>();
                Boolean remain = true;
                int m = 0;
                while (remain && m < 30)
                {
                    try
                    {
                        byte[] packetData;
                        remain = parseQueue.TryDequeue(out packetData);
                        if (remain)
                        {
                            result.Add(packetData);
                        }
                    }
                    catch (Exception ex)
                    {
                        logger.Error(ex.Message, ex);
                    }
                    m++;
                }
                if (result.Count > 0)
                {
                    //logger.Info("ParseThreadFunc->放入多线程解析，个数=" + result.Count);
                    ThreadPool.QueueUserWorkItem(new WaitCallback(ParseData), result);
                }
                Thread.Sleep(10);
            }
        }
        private void ParseData(object state)
        {
            List<byte[]> result = (List<byte[]>)state;
            foreach (byte[] bytes in result)
            {
                try
                {
                    BdPacketData bdPacketData = ReceiveProtocol.DecodeProtocol(bytes);
                    //通知
                    PacketNotify(bdPacketData);
                    
                }
                catch (Exception ex)
                {
                    logger.Error(ex.Message, ex);
                }
            }
        }

    }
}
